package com.gitee.melin.bee.core.jdbc.dialect.impl;

import com.gitee.melin.bee.core.jdbc.ConnectionFactory;
import com.gitee.melin.bee.core.jdbc.ConnectionProxy;
import com.gitee.melin.bee.core.jdbc.DriverConnectionFactory;
import com.gitee.melin.bee.core.jdbc.JdbcInvocationHandler;
import com.gitee.melin.bee.core.jdbc.NamedParameterStatement;
import com.gitee.melin.bee.core.jdbc.QueryResult;
import com.gitee.melin.bee.core.jdbc.dialect.AbstractJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.JdbcDialect;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.enums.FileFormat;
import com.gitee.melin.bee.core.jdbc.relational.BeeTableType;
import com.gitee.melin.bee.core.jdbc.relational.ConnectionInfo;
import com.gitee.melin.bee.core.jdbc.relational.DatabaseVersion;
import com.gitee.melin.bee.core.jdbc.relational.MetaColumn;
import com.gitee.melin.bee.core.jdbc.relational.MetaDataSource;
import com.gitee.melin.bee.core.jdbc.relational.MetaTable;
import com.gitee.melin.bee.core.jdbc.relational.SchemaIdentifier;
import com.gitee.melin.bee.core.jdbc.relational.SimpleMetaTable;
import com.gitee.melin.bee.util.JdbcUtils;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.mysql.cj.jdbc.Driver;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class IcebergJdbcDialect implements JdbcDialect {

    private static final Logger LOG = LoggerFactory.getLogger(AbstractJdbcDialect.class);

    private static final String QUERY_SCHEMA_SQL =
            "select distinct table_namespace as database_name from iceberg_tables";

    private ConnectionInfo connectionInfo;

    private ConnectionFactory connectionFactory;

    private Connection connection;

    public IcebergJdbcDialect(ConnectionInfo connectionInfo) {
        if (connectionInfo == null) {
            return;
        }

        this.connectionInfo = connectionInfo;

        Map<String, String> properties = connectionInfo.getProperties();
        if (properties == null) {
            properties = Maps.newHashMap();
        }

        connectionFactory = createConnectionFactory(properties);
    }

    public ConnectionFactory createConnectionFactory(Map<String, String> properties) {
        try {
            return new DriverConnectionFactory(new Driver(), connectionInfo.getUrl(), connectionInfo, properties, null);
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    public IcebergJdbcDialect(Connection connection) {
        this.connection = connection;
    }

    @Override
    public DataSourceType getDataSourceType() {
        return DataSourceType.ICEBRG;
    }

    private Connection getConnection() throws SQLException {
        if (connectionFactory != null) {
            return connectionFactory.openConnection();
        } else {
            return createConnectionProxy(connection); // 避免关闭连接
        }
    }

    private Connection createConnectionProxy(Connection con) {
        return (Connection) Proxy.newProxyInstance(
                ConnectionProxy.class.getClassLoader(),
                new Class<?>[] {ConnectionProxy.class},
                new JdbcInvocationHandler(con));
    }

    @Override
    public MetaDataSource testConnection(ConnectionInfo connectionInfo) {
        try (Connection connection = DriverManager.getConnection(
                connectionInfo.getUrl(), connectionInfo.getUsername(), connectionInfo.getPassword())) {

            DatabaseVersion version = getDatabaseVersion(connection);
            return new MetaDataSource(connection.getMetaData(), version);
        } catch (SQLException e) {
            throw new RuntimeException("Test connection failed: " + e.getMessage(), e);
        }
    }

    private DatabaseVersion getDatabaseVersion(Connection connection) {
        try {
            DatabaseMetaData metaData = connection.getMetaData();

            try {
                return new DatabaseVersion(metaData.getDatabaseProductVersion());
            } catch (Throwable e) {
                try {
                    int major = metaData.getDatabaseMajorVersion();
                    int minor = metaData.getDatabaseMinorVersion();
                    return new DatabaseVersion(major, minor, 0);
                } catch (IllegalArgumentException e1) {
                    LOG.debug("Can't determine database version. Use default");
                    return new DatabaseVersion(0, 0, 0);
                }
            }
        } catch (SQLException e) {
            throw new RuntimeException("Test connection failed: " + e.getMessage(), e);
        }
    }

    @Override
    public DatabaseVersion getDatabaseVersion() {
        throw new UnsupportedOperationException();
    }

    @Override
    public List<String> getDatabases() {
        return this.getSchemas();
    }

    @Override
    public List<String> getSchemas() {
        ResultSet resultSet = null;
        try (Connection connection = this.getConnection();
                PreparedStatement preparedStatement = connection.prepareStatement(QUERY_SCHEMA_SQL)) {

            if (connectionInfo != null && connectionInfo.getQueryTimeout() != null) {
                preparedStatement.setQueryTimeout(connectionInfo.getQueryTimeout());
            }

            resultSet = preparedStatement.executeQuery();
            List<String> schemas = Lists.newArrayList();
            while (resultSet.next()) {
                String schemaName = resultSet.getString("database_name");
                schemas.add(schemaName);
            }

            return schemas;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error", e);
        } finally {
            JdbcUtils.closeResultSet(resultSet);
        }
    }

    private String getSelectTablesSql(String tableName) {
        String sql =
                "select table_namespace as database_name, table_name from iceberg_tables where table_namespace = :schemaName";
        if (StringUtils.isNotBlank(tableName)) {
            sql += " and table_name = :tableName";
        }
        return sql;
    }

    private List<MetaTable> getSchemaTables(String schemaName, String tableName) {
        ResultSet resultSet = null;
        final String sql = getSelectTablesSql(tableName);
        try (Connection connection = this.getConnection();
                NamedParameterStatement preparedStatement = new NamedParameterStatement(connection, sql)) {
            preparedStatement.setString("schemaName", schemaName);

            if (connectionInfo != null && connectionInfo.getQueryTimeout() != null) {
                preparedStatement.setQueryTimeout(connectionInfo.getQueryTimeout());
            }

            preparedStatement.setString("schemaName", schemaName);
            if (StringUtils.isNotBlank(tableName)) {
                preparedStatement.setString("tableName", tableName);
            }

            resultSet = preparedStatement.executeQuery();
            List<MetaTable> schemas = Lists.newArrayList();
            while (resultSet.next()) {
                String dbName = resultSet.getString("database_name");
                String tblName = resultSet.getString("table_name");
                MetaTable metaTable = new MetaTable(dbName, tblName);
                metaTable.setTableType(BeeTableType.TABLE);
                metaTable.setFileFormat(FileFormat.PAIMON);
                schemas.add(metaTable);
            }

            return schemas;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error", e);
        } finally {
            JdbcUtils.closeResultSet(resultSet);
        }
    }

    @Override
    public MetaTable getSchemaTable(String schemaName, String tableName, boolean containColumn) {
        List<MetaTable> tables = this.getSchemaTables(schemaName, tableName);
        if (tables.isEmpty()) {
            return null;
        } else if (tables.size() > 1) {
            String msg = String.format("table %s.%s has more than one", schemaName, tableName);
            throw new RuntimeException(msg);
        } else {
            MetaTable tableInfo = tables.get(0);
            if (containColumn) {
                List<MetaColumn> columns = this.getSchemaColumns(schemaName, tableName);
                tableInfo.setColumns(columns);
            }
            return tableInfo;
        }
    }

    @Override
    public List<String> getTableNames(String schemaName) {
        ResultSet resultSet = null;
        final String sql = getSelectTablesSql(null);
        try (Connection connection = this.getConnection();
                NamedParameterStatement preparedStatement = new NamedParameterStatement(connection, sql)) {
            preparedStatement.setString("schemaName", schemaName);

            if (connectionInfo != null && connectionInfo.getQueryTimeout() != null) {
                preparedStatement.setQueryTimeout(connectionInfo.getQueryTimeout());
            }

            resultSet = preparedStatement.executeQuery();
            List<String> schemas = Lists.newArrayList();
            while (resultSet.next()) {
                String tblName = resultSet.getString("table_name");
                schemas.add(tblName);
            }

            return schemas;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error", e);
        } finally {
            JdbcUtils.closeResultSet(resultSet);
        }
    }

    @Override
    public List<SimpleMetaTable> getSimpleTables(String schemaName) {
        return this.getSchemaTables(schemaName, null).stream()
                .map(meta -> {
                    SimpleMetaTable metaTable = new SimpleMetaTable();
                    metaTable.setTableName(meta.getTableName());
                    metaTable.setTableType(BeeTableType.TABLE);
                    return metaTable;
                })
                .collect(Collectors.toList());
    }

    @Override
    public List<MetaTable> getSchemaTables(String schemaName) {
        return this.getSchemaTables(schemaName, null);
    }

    @Override
    public LinkedHashSet<String> getPrimaryKeys(String schemaName, String tableName) {
        return null;
    }

    @Override
    public List<MetaColumn> getSchemaColumns(String schemaName, String tableName) {
        return Collections.emptyList();
    }

    @Override
    public String getCreateTableScript(BeeTableType type, String schemaName, String tableName) {
        return "";
    }

    @Override
    public Statement getQueryStatement() {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String schema, String sql) {
        return null;
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String sql, int maxRecords) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String schema, String sql, int maxRecords) {
        throw new UnsupportedOperationException();
    }

    @Override
    public QueryResult query(String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public QueryResult query(String schema, String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public QueryResult query(String sql, int maxRecords) {
        throw new UnsupportedOperationException();
    }

    @Override
    public QueryResult query(String schema, String sql, int maxRecords) {
        throw new UnsupportedOperationException();
    }

    @Override
    public long queryCount(String sql, Object... params) {
        throw new UnsupportedOperationException();
    }

    @Override
    public List<Map<String, Object>> queryForList(String sql, Object... params) {
        return Collections.emptyList();
    }

    @Override
    public Boolean execute(String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Boolean execute(String schema, String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<Boolean> asyncExecute(String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<Boolean> asyncExecute(String schema, String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Integer executeUpdate(String sql, Object... params) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Integer executeUpdate(SchemaIdentifier schema, String sql, Object... params) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<Integer> asyncExecuteUpdate(String sql, Object... params) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<Integer> asyncExecuteUpdate(SchemaIdentifier schema, String sql, Object... params) {
        throw new UnsupportedOperationException();
    }

    @Override
    public void close() throws Exception {
        if (connectionFactory != null) {
            try {
                connectionFactory.close();
            } catch (Exception e) {
            }
        }
    }
}
